package com.xxljob.demo.core.thread;


import com.xxljob.demo.core.conf.XxlJobAdminConfig;
import com.xxljob.demo.core.cron.CronExpression;
import com.xxljob.demo.core.model.XxlJobInfo;
import com.xxljob.demo.core.trigger.TriggerTypeEnum;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

import java.sql.Connection;
import java.sql.PreparedStatement;
import java.sql.SQLException;
import java.text.ParseException;
import java.util.*;
import java.util.concurrent.ConcurrentHashMap;
import java.util.concurrent.TimeUnit;

/**
 * @Author：YH
 * @Description：调度器 真正的对任务进行调度触发的地方
 * @Date:Created in 2022/5/19 22:13
 */
public class JobScheduleHelper {


    private static final Logger LOGGER = LoggerFactory.getLogger(JobScheduleHelper.class);

    private static JobScheduleHelper helper = new JobScheduleHelper();

    public static JobScheduleHelper getInstance() {
        return helper;
    }

    public static final long PRE_READ_MS = 5000;
    private Thread scheduleThread;
    private Thread ringThread;

    private volatile boolean scheduleThreadToStop = false;
    private volatile boolean ringThreadToStop = false;

    private volatile static Map<Integer, List<Integer>> ringData = new ConcurrentHashMap<>();

    public void start() {
        scheduleThread = new Thread(new Runnable() {

            @Override
            public void run() {

                try {
                    //5秒后的整秒   当前毫秒%1000 是取余得到下次整秒的事件距离
                    TimeUnit.MILLISECONDS.sleep(5000 - System.currentTimeMillis() % 1000);
                } catch (InterruptedException e) {
                    if (!scheduleThreadToStop) {
                        LOGGER.error(e.getMessage(), e);
                    }
                }

                LOGGER.info(">>>>>>>>> JobScheduleHelper调度线程启动 init xxl-job admin scheduler success.");

                // pre-read count: treadpool-size * trigger-qps (each trigger cost 50ms, qps = 1000/50 = 20)
                int preReadCount =
                        (XxlJobAdminConfig.getAdminConfig().getTriggerPoolFastMax() + XxlJobAdminConfig.getAdminConfig().getTriggerPoolSlowMax()) * 20;
                while (!scheduleThreadToStop) {
                    // Scan Job
                    long start = System.currentTimeMillis();
                    Connection conn = null;
                    Boolean connAutoCommit = null;
                    PreparedStatement preparedStatement = null;
                    boolean preReadSuc = true;

                    try {
                        conn = XxlJobAdminConfig.getAdminConfig().getDataSource().getConnection();
                        connAutoCommit = conn.getAutoCommit();
                        conn.setAutoCommit(false);

                        preparedStatement = conn.prepareStatement("select * from xxl_job_lock where lock_name = " +
                                "'schedule_lock' for update");
                        preparedStatement.execute();
                        // 1、pre read
                        long nowTime = System.currentTimeMillis();
                        //查找下次的调度时间
                        List<XxlJobInfo> scheduleList =
                                XxlJobAdminConfig.getAdminConfig().getXxlJobInfoDao().scheduleJobQuery(nowTime + PRE_READ_MS, preReadCount);
                        if (Objects.nonNull(scheduleList) && scheduleList.size() > 0) {
                            // 2、push time-ring
                            for (XxlJobInfo info : scheduleList) {
                                //若果当前时间 > 该任务下次执行时间 + 5秒  那说明这个任务调度超时了 当前时间超过了本该执行时间+5秒
                                if (nowTime > info.getTriggerNextTime() + PRE_READ_MS) {
                                    LOGGER.warn(">>>>>>>>>>>>xxl-job,shedule misfire,jobID=" + info.getId());
                                    // fresh next
                                    refreshNextValidTime(info, new Date());
                                    JobTriggerPoolHelper.trigger(info.getId(), TriggerTypeEnum.CRON, -1, null, null,
                                            null);

                                } else if (nowTime > info.getTriggerNextTime()) {
                                    // 1、trigger
                                    LOGGER.debug(">>>>>>>>>>> 调度中心准备触发 xxl-job, schedule push trigger : jobId = " + info.getId());
                                    JobTriggerPoolHelper.trigger(info.getId(), TriggerTypeEnum.CRON, -1, null, null,
                                            null);
                                    // 2、fresh next

                                    // next-trigger-time in 5s, pre-read again
                                    if (info.getTriggerStatus() == 1 && nowTime + PRE_READ_MS > info.getTriggerNextTime()) {

                                        // 1、make ring second
                                        int ringSecond = (int) ((info.getTriggerNextTime() / 1000) % 60);

                                        // 2、push time ring
                                        pushTimeRing(ringSecond, info.getId());

                                        // 3、fresh next
                                        refreshNextValidTime(info, new Date(info.getTriggerNextTime()));
                                    } else {
                                        // 2.3、trigger-pre-read：time-ring trigger && make next-trigger-time

                                        // 1、make ring second
                                        int ringSecond = (int) ((info.getTriggerNextTime() / 1000) % 60);

                                        // 2、push time ring
                                        pushTimeRing(ringSecond, info.getId());

                                        // 3、fresh next
                                        refreshNextValidTime(info, new Date(info.getTriggerNextTime()));

                                    }
                                } else {
                                    LOGGER.debug(">>>>>>>>>>> 调度中心时间还未到 xxl-job, schedule push trigger : jobId = " + info.getId());
                                    // 如果不是以上两种情况，那这个任务应该是没到时间的  加入时间轮等待执行
                                    JobTriggerPoolHelper.trigger(info.getId(), TriggerTypeEnum.CRON, -1, null, null,
                                            null);

                                    // 2、fresh next
                                    //refreshNextValidTime(info, new Date());

                                }
                            }

                            // 3、update trigger info
                            for (XxlJobInfo jobInfo : scheduleList) {
                                XxlJobAdminConfig.getAdminConfig().getXxlJobInfoDao().scheduleUpdate(jobInfo);
                            }
                        }
                    } catch (Exception e) {
                        if (!scheduleThreadToStop) {
                            LOGGER.error(">>>>>>>>>>> xxl-job, JobScheduleHelper#scheduleThread error:{}", e);
                        }
                    } finally {
                        if (Objects.nonNull(conn)) {
                            try {
                                conn.commit();
                            } catch (SQLException e) {
                                if (!scheduleThreadToStop) {
                                    LOGGER.error(e.getMessage(), e);
                                }
                            }

                            try {
                                conn.setAutoCommit(connAutoCommit);
                            } catch (SQLException e) {
                                if (!scheduleThreadToStop) {
                                    LOGGER.error(e.getMessage(), e);
                                }
                            }
                            try {
                                conn.close();
                            } catch (SQLException e) {
                                if (!scheduleThreadToStop) {
                                    LOGGER.error(e.getMessage(), e);
                                }
                            }
                        }
                        // close PreparedStatement
                        if (null != preparedStatement) {
                            try {
                                preparedStatement.close();
                            } catch (SQLException e) {
                                if (!scheduleThreadToStop) {
                                    LOGGER.error(e.getMessage(), e);
                                }
                            }
                        }
                    }

                    long cost = System.currentTimeMillis() - start;
                    // Wait seconds, align second
                    if (cost < 1000) {  // scan-overtime, not wait
                        try {
                            // pre-read period: success > scan each second; fail > skip this period;
                            TimeUnit.MILLISECONDS.sleep((preReadSuc ? 1000 : PRE_READ_MS) - System.currentTimeMillis() % 1000);
                        } catch (InterruptedException e) {
                            if (!scheduleThreadToStop) {
                                LOGGER.error(e.getMessage(), e);
                            }
                        }
                    }
                }
            }
        });
        scheduleThread.setDaemon(true);
        scheduleThread.setName("xxl-job, admin JobScheduleHelper scheduleThread");
        scheduleThread.start();

        ringThread = new Thread(new Runnable() {
            @Override
            public void run() {
                LOGGER.info("xxl-job, admin JobScheduleHelper ringThread");


            }
        });
        ringThread.setDaemon(true);
        ringThread.setName("xxl-job, admin JobScheduleHelper ringThread");
        ringThread.start();
    }

    private void pushTimeRing(int ringSecond, int jobId) {
        // push async ring
        List<Integer> ringItemData = ringData.get(ringSecond);
        if (ringItemData == null) {
            ringItemData = new ArrayList<Integer>();
            ringData.put(ringSecond, ringItemData);
        }
        ringItemData.add(jobId);
        LOGGER.debug(">>>>>>>>>>> xxl-job, schedule push time-ring : " + ringSecond + " = " + Arrays.asList(ringItemData));
    }

    public void toStop() {
        // 1、stop schedule
        scheduleThreadToStop = true;
        try {
            TimeUnit.SECONDS.sleep(1);
        } catch (InterruptedException e) {
            LOGGER.error(e.getMessage(), e);
        }
        if (scheduleThread.getState() != Thread.State.TERMINATED) {
            // interrupt and wait
            scheduleThread.interrupt();
            try {
                scheduleThread.join();
            } catch (InterruptedException e) {
                LOGGER.error(e.getMessage(), e);
            }
        }
        // if has ring data
        boolean hasRingData = false;
        if (!ringData.isEmpty()) {
            for (int second : ringData.keySet()) {
                List<Integer> tmpData = ringData.get(second);
                if (tmpData != null && tmpData.size() > 0) {
                    hasRingData = true;
                    break;
                }
            }
        }
        if (hasRingData) {
            try {
                TimeUnit.SECONDS.sleep(8);
            } catch (InterruptedException e) {
                LOGGER.error(e.getMessage(), e);
            }
        }

        // stop ring (wait job-in-memory stop)
        ringThreadToStop = true;
        try {
            TimeUnit.SECONDS.sleep(1);
        } catch (InterruptedException e) {
            LOGGER.error(e.getMessage(), e);
        }
        if (ringThread.getState() != Thread.State.TERMINATED) {
            // interrupt and wait
            ringThread.interrupt();
            try {
                ringThread.join();
            } catch (InterruptedException e) {
                LOGGER.error(e.getMessage(), e);
            }
        }
        LOGGER.info(">>>>>>>>>>> xxl-job, JobScheduleHelper stop");
    }


    /**
     * @Description:修改任务的触发时间和触发状态
     **/
    private void refreshNextValidTime(XxlJobInfo jobInfo, Date fromtime) throws ParseException {
        //下次执行的时间
        Date nextValidTime = new CronExpression(jobInfo.getJobCron()).getNextValidTimeAfter(fromtime);
        if (Objects.nonNull(nextValidTime)) {
            jobInfo.setTriggerLastTime(jobInfo.getTriggerNextTime());
            jobInfo.setTriggerNextTime(nextValidTime.getTime());
        } else {
            //设置为未启动触发
            jobInfo.setTriggerStatus(0);
            jobInfo.setTriggerLastTime(0);
            jobInfo.setTriggerNextTime(0);
        }
    }
}

